package menu ;

import java.util.* ;
import java.util.regex.* ;
import java.net.* ;
import java.time.* ;

import io.github.cdimascio.dotenv.Dotenv;
import org.apache.logging.log4j.Logger ;
import org.apache.logging.log4j.LogManager ;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.GenericData;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.CqlSessionBuilder ;
import com.datastax.oss.driver.api.core.cql.Row;

import com.fasterxml.jackson.databind.ObjectMapper;

public class DashboardMenu {
    private static final Logger LOGGER = LogManager.getLogger(LogManager.ROOT_LOGGER_NAME) ;

    private static String _resolveEnv(Dotenv dotenv, String name) {
        String v = dotenv.get(name);
        if (v == null)
            return null;
        Matcher p = Pattern.compile("\\$\\{([^}]*)\\}").matcher(v);
        
        String ret = v;
        while (p.find()) {
            String subVarValue = _resolveEnv(dotenv, p.group(1));
            if (subVarValue == null)
                return null;
            ret = v.replace(p.group(0), subVarValue);
        }
        return ret ;
    }
    
    public static String resolveEnv(Dotenv dotenv, String name, String defaultValue) {
        String resolveValue = _resolveEnv(dotenv, name);
        return resolveValue != null ? resolveValue : defaultValue;
    }

    public static String resolveMyIp(String nodes) throws Exception {
        for (NetworkInterface ni : Collections.list(NetworkInterface.getNetworkInterfaces())) {
            for (InetAddress addr : Collections.list(ni.getInetAddresses())) {
                String currentIp = addr.getHostAddress();
                if (nodes.contains(currentIp)) {
                    return currentIp;
                }
            }
        }
        return null;
    }
    
    public static Set<String> loadDb(CqlSession cqlSession, String cql) {
        Set<String> results = new HashSet<String>() ;
        for(Row row: cqlSession.execute(cql)) {
            results.add(row.getObject(0).toString()) ;
        }
        return results ;
    }

    public static CqlSession mkCqlSession(String cassandras) {
        CqlSessionBuilder sqlSessionBuilder = CqlSession.builder() ;
        for(String node: cassandras.split(",")) {
            String[] nodeParts = node.split(":") ;
            String host = nodeParts[0] ;
            int port = Integer.parseInt(nodeParts[1]) ;
            sqlSessionBuilder.addContactPoint(
                new InetSocketAddress(host, port)
            ) ;
        }
        return sqlSessionBuilder
            .withLocalDatacenter("datacenter1")
            .build();
    }
    public static KafkaConsumer<byte[], GenericData.Array<GenericRecord>> mkKafkaConsumer(String kafkas, String kafkaSrs) {
        Properties inProps = new Properties();
        inProps.put("bootstrap.servers", kafkas);
        inProps.put("auto.offset.reset", "earliest");
        inProps.put("group.id", "enum-server-java__menu__dashboard");
        inProps.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        inProps.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
        inProps.put("schema.registry.url", kafkaSrs);
        inProps.put("enable.auto.commit", "false") ;

        return new KafkaConsumer<byte[], GenericData.Array<GenericRecord>>(inProps) ;
    }
    public static void main(String[] args) throws Exception {
        Dotenv dotenv = Dotenv.configure().directory("meta-lib-static/env/.env").load();
        
        String hosts= resolveEnv(dotenv, "ENUM_SERVER_HOSTS__STREAMING", "127.0.0.1");

        String kafkas = resolveEnv(dotenv, "ENUM_SERVER_KAFKAS", "127.0.0.1:9092") ;
        String kafka_srs = resolveEnv(dotenv, "ENUM_SERVER_KAFKA_SRS", "http://localhost:8081") ;
        String cassandras = resolveEnv(dotenv, "ENUM_SERVER_CASSANDRAS__STREAMING", "") ;

        String buffer_schema = resolveEnv(dotenv, "ENUM_SERVER_BUFFER_SCHEMA", "data_buffer_dev") ;
        String channel_schema = resolveEnv(dotenv, "ENUM_SERVER_CHANNEL_SCHEMA", "data_channel_dev") ;

        String inTopic = String.format("%s_dc_sdk_push", buffer_schema) ;

        String cql = "SELECT JSON * FROM " + channel_schema + ".dashboard_menu" ;
        CqlSession cqlSession = mkCqlSession(cassandras) ;
        Set<String> state = loadDb(cqlSession, cql) ;

        // System.out.println("state:" + state) ;
        KafkaConsumer<byte[], GenericData.Array<GenericRecord>> consumer = mkKafkaConsumer(kafkas, kafka_srs) ;
        consumer.subscribe(Arrays.asList(inTopic));

        while(true) {
            for (ConsumerRecord<byte[], GenericData.Array<GenericRecord>> messages : consumer.poll(Duration.ofMillis(1000))) {
                for (GenericRecord sdkRecord : messages.value()) {
                    GenericRecord avroHeader = (GenericRecord) sdkRecord.get("common") ;
                    GenericRecord basicInfo = (GenericRecord) avroHeader.get("basic") ;

                    String tenant_id = String.valueOf(basicInfo.get("tenant_id")) ;
                    String project_id = String.valueOf(basicInfo.get("project_id")) ;
                    String app_id = String.valueOf(basicInfo.get("app_id")) ;
                    String channel = String.valueOf((basicInfo.get("market_name") != null ? basicInfo.get("market_name") : basicInfo.get("app_package_id"))) ;
                    if (channel == null) channel = "null" ;
                    String version = String.valueOf(basicInfo.get("app_version")) ;
                    if (version == null) version = "null" ;

                    Map<String, String> m = new HashMap<String, String>() ;
                    m.put("tenant_id", tenant_id) ;
                    m.put("project_id", project_id) ;
                    m.put("app_id", app_id) ;
                    m.put("channel", channel) ;
                    m.put("version", version) ;

                    String json = new ObjectMapper().writeValueAsString(m) ;
                    if (state.contains(json)) continue ;
                    state.add(json) ;
                    
                    cqlSession.execute("INSERT INTO " + channel_schema + ".dashboard_menu JSON ?", json) ;
                }
            }
            consumer.commitSync(Duration.ofMillis(5000));
        }
    }
}
